#ifndef AVALON_SERVICEPROTOCOLSERVER_HPP
#define AVALON_SERVICEPROTOCOLSERVER_HPP
#include <boost/call_traits.hpp>
#include <boost/preprocessor/repetition/enum_params.hpp>
#include <boost/preprocessor/repetition/enum_binary_params.hpp>
#include <boost/preprocessor/repetition/repeat.hpp>
#include <boost/preprocessor/iteration/local.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/recursive_mutex.hpp>
#include "Avalon/Base/Algorithm.hpp"
#include "Avalon/IO/AcceptState.hpp"
#include "Avalon/IO/ConnectException.hpp"
#include "Avalon/IO/ConnectionServer.hpp"
#include "Avalon/Serialization/DataShuttleFactory.hpp"
#include "Avalon/Services/Services.hpp"
#include "Avalon/Services/ServiceProtocol.hpp"
#include "Avalon/SignalHandling/TaskSignalHandler.hpp"

namespace Avalon {
namespace Services {

  /*! \class ServiceProtocolServer
      \brief Implements the server side of the service protocol.
      \tparam ChannelType The type of Channel accepted by the server.
   */
  template<typename ChannelType>
  class ServiceProtocolServer {
    public:

      //! Called when a Channel is accepted.
      /*!
        \param channel The Channel that was accepted.
      */
      typedef typename IO::ConnectionServer<ChannelType>::AcceptSlot AcceptSlot;

      //! Called when a Channel is closed.
      /*!
        \param channel The Channel that closed.
      */
      typedef typename IO::ConnectionServer<ChannelType>::ChannelClosedSlot
        ChannelClosedSlot;

      //! Called when the ServerConnection is closed.
      typedef typename IO::ConnectionServer<
        ChannelType>::ServerConnectionClosedSlot ServerConnectionClosedSlot;

      //! Constructs a ServiceProtocolServer.
      ServiceProtocolServer();

      ~ServiceProtocolServer();

      //! Initializes the ServiceProtocolServer.
      /*!
        \param serverConnection The ServerConnection accepting
                                ClientConnections.
        \param dataShuttleFactory The type of Message Serialization to use.
        \param acceptSlot The slot to call when a Channel is accepted.
        \param channelClosedSlot The slot to call when a Channel is closed.
        \param serverConnectionClosedSlot The slot to call when the
                                          ServerConnection is closed.
        \param threadPool The ThreadPool to execute asynchronous tasks.
      */
      void Initialize(IO::ServerConnection* serverConnection,
        Serialization::DataShuttleFactory* dataShuttleFactory,
        const AcceptSlot& acceptSlot,
        const ChannelClosedSlot& channelClosedSlot,
        const ServerConnectionClosedSlot& serverConnectionClosedSlot,
        Threading::ThreadPool& threadPool);

      //! Returns <code>true</code> iff this server is open.
      bool IsOpen() const;

      //! Opens this server.
      void Open();

      //! Closes this server.
      void Close();

      //! Sets the request handler for a service.
      /*!
        \tparam Service The Service being handled.
        \param slot The slot called to handle the <i>Service</i>.
      */
      template<typename Service>
      void SetHandler(const typename Service::Request::Slot& slot);

      //! Marks an task as pending, queuing it for later processing.
      /*!
        \param channel The Channel that made the request.
        \param task The task to enqueue.
      */
      void AddPendingTask(ChannelType* channel,
        const boost::function<void ()>& task);

      //! Sends a Message to a list of Channels.
      /*!
        \param channels The Channels to send the Message to.
        \param message The Message to send.
      */
      void SendMessage(const std::vector<ChannelType*>& channels,
        const Message& message);

    private:
      mutable boost::recursive_mutex m_mutex;
      IO::ConnectionServer<ChannelType> m_connectionServer;
      boost::scoped_ptr<Serialization::DataShuttleFactory> m_dataShuttleFactory;
      bool m_isOpen;
      bool m_isClosing;
      AcceptSlot m_acceptSlot;
      ChannelClosedSlot m_channelClosedSlot;
      std::map<const std::type_info*, boost::function_base*> m_serviceSlots;
      mutable boost::condition_variable_any m_closedCondition;
      SignalHandling::TaskSignalHandler m_signalHandler;
      Threading::ThreadPool* m_threadPool;

      void WaitForClose(boost::unique_lock<boost::recursive_mutex>& lock) const;
      void CloseResources();
      void CloseChannel(ChannelType* channel);
      void OnChannelAccepted(ChannelType* channel);
      void OnChannelClosed(ChannelType* channel);
      void OnChannelTaskComplete(ChannelType* channel);

      template<typename Service, int Arity =
        boost::function_traits<Service::Request::Slot>::arity - 1>
      struct GetWrapperSlot {};

      #define DECLARE_PARAMETER(z, n, q)                                       \
        BOOST_PP_COMMA_IF(n) typename boost::call_traits<                      \
          typename Service::Parameter##n::Type>::param_type a##n

      #define BOOST_PP_LOCAL_MACRO(n)                                          \
      template<typename Service>                                               \
      struct GetWrapperSlot<Service, n> {                                      \
        static typename Service::Request::Slot* Invoke(                        \
          ServiceProtocolServer* server, const typename Service::Request::Slot&\
          slot);                                                               \
      };                                                                       \
                                                                               \
      template<typename Service>                                               \
      void Slot##n(const RequestToken<Service>& request,                       \
        const typename Service::Request::Slot& slot BOOST_PP_COMMA_IF(n)       \
        BOOST_PP_REPEAT(n, DECLARE_PARAMETER,));

      #define BOOST_PP_LOCAL_LIMITS (0, AVALON_SERVICE_PARAMETERS)
      #include BOOST_PP_LOCAL_ITERATE()
      #undef DECLARE_PARAMETER
  };

  template<typename ChannelType>
  ServiceProtocolServer<ChannelType>::ServiceProtocolServer() {}

  template<typename ChannelType>
  ServiceProtocolServer<ChannelType>::~ServiceProtocolServer() {
    assert(!m_isOpen);
    assert(!m_isClosing);
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::Initialize(
      IO::ServerConnection* serverConnection,
      Serialization::DataShuttleFactory* dataShuttleFactory,
      const AcceptSlot& acceptSlot,
      const ChannelClosedSlot& channelClosedSlot,
      const ServerConnectionClosedSlot& serverConnectionClosedSlot,
      Threading::ThreadPool& threadPool) {
    m_connectionServer.Initialize(serverConnection,
      boost::bind(&ServiceProtocolServer::OnChannelAccepted, this, _1),
      boost::bind(&ServiceProtocolServer::OnChannelClosed, this, _1),
      serverConnectionClosedSlot);
    m_dataShuttleFactory.reset(dataShuttleFactory);
    m_isOpen = false;
    m_isClosing = false;
    m_acceptSlot = acceptSlot;
    m_channelClosedSlot = channelClosedSlot;
    Clear(m_serviceSlots);
    m_signalHandler.Initialize(threadPool);
    m_threadPool = &threadPool;
  }

  template<typename ChannelType>
  bool ServiceProtocolServer<ChannelType>::IsOpen() const {
    boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
    return m_isOpen;
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::Open() {
    boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
    if(m_isOpen) {
      return;
    }
    m_connectionServer.Open();
    m_isOpen = true;
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::Close() {
    boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
    if(!m_isOpen) {
      return;
    }
    CloseResources();
    WaitForClose(lock);
  }

  template<typename ChannelType>
  template<typename Service>
  void ServiceProtocolServer<ChannelType>::SetHandler(const
      typename Service::Request::Slot& slot) {
    typename Service::Request::Slot* wrapperSlot =
      GetWrapperSlot<Service>::Invoke(this, slot);
    assert(m_serviceSlots.find(&typeid(Service)) == m_serviceSlots.end());
    m_serviceSlots.insert(std::make_pair(&typeid(Service), wrapperSlot));
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::AddPendingTask(ChannelType* channel,
      const boost::function<void ()>& task) {
    channel->GetTaskQueue().AddTask(task);
    if(channel->GetTaskQueue().GetRunningTaskCount() == 0) {
      channel->GetTaskQueue().StartNextTask(Store(
        channel->GetLastTaskResult()));
    }
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::SendMessage(
      const std::vector<ChannelType*>& channels, const Message& message) {
    for(typename std::vector<ChannelType*>::const_iterator i = channels.begin();
        i != channels.end(); ++i) {
      (*i)->SendMessage(message);
    }
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::CloseResources() {
    m_isClosing = true;
    m_connectionManager.Close();
    m_isClosing = false;
    m_isOpen = false;
    m_closedCondition.notify_one();
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::CloseChannel(ChannelType* channel) {

    // TODO
    channel->GetServiceProtocol().Stop();
    m_channelClosedSlot(channel);
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::WaitForClose(
      boost::unique_lock<boost::recursive_mutex>& lock) const {
    if(m_isOpen || m_isClosing) {
      m_closedCondition.wait(lock);
    }
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::OnChannelClosed(
      ChannelType* channel) {
    boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
    channel->GetConnectionState().SetClosed();
    if(channel->GetTaskQueue().IsEmpty()) {
      CloseChannel(channel);
    }
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::OnChannelAccepted(
      ChannelType* channel) {
    boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
    channel->GetConnectionState().Initialize(true);
    channel->GetTaskQueue().Initialize(*m_threadPool);
    Threading::Async<void>::State taskState;
    channel->GetLastTaskResult().SetFinishedSlot(
      m_signalHandler.template GetSlot<Threading::Async<void>::FinishedSlot>(
      boost::bind(&ServiceProtocolServer::OnChannelTaskComplete, this,
      channel)), Store(taskState));
    channel->GetServiceProtocol().Initialize(channel,
      m_dataShuttleFactory->CreateSerializer(),
      m_dataShuttleFactory->CreateDeserializer(),
      &m_serviceSlots, *m_threadPool);
    channel->GetServiceProtocol().Start();
    m_acceptSlot(channel);
  }

  template<typename ChannelType>
  void ServiceProtocolServer<ChannelType>::OnChannelTaskComplete(
      ChannelType* channel) {
    boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
    channel->GetTaskQueue().MarkTaskAsFinished(&channel->GetLastTaskResult());
    if(channel->GetTaskQueue().GetRunningTaskCount() == 0) {
      if(channel->GetTaskQueue().GetPendingTaskCount() != 0) {
        channel->GetLastTaskResult().ResetResult();
        channel->GetTaskQueue().StartNextTask(
          Store(channel->GetLastTaskResult()));
      } else if(!channel->GetConnectionState().IsOpen()) {
        CloseChannel(channel);
      }
    }
  }

  #define DECLARE_PARAMETER(z, n, q)                                           \
    BOOST_PP_COMMA_IF(n) typename boost::call_traits<                          \
      typename Service::Parameter##n::Type>::param_type a##n

  #define PLACEHOLDERS(z, n, q)                                                \
    BOOST_PP_COMMA_IF(n) boost::arg<n + 2>()

  #define LIST_PARAMETERS(z, n, q)                                             \
    BOOST_PP_COMMA_IF(n) a##n

  #define BOOST_PP_LOCAL_MACRO(n)                                              \
  template<typename ChannelType>                                               \
  template<typename Service>                                                   \
  typename Service::Request::Slot*                                             \
      ServiceProtocolServer<ChannelType>::GetWrapperSlot<Service, n>::Invoke(  \
      ServiceProtocolServer* server, const typename                            \
      Service::Request::Slot& slot) {                                          \
    return new typename Service::Request::Slot(boost::bind(                    \
      &ServiceProtocolServer::Slot##n<Service>, server, boost::arg<1>(), slot  \
      BOOST_PP_COMMA_IF(n) BOOST_PP_REPEAT(n, PLACEHOLDERS,)));                \
  }                                                                            \
                                                                               \
  template<typename ChannelType>                                               \
  template<typename Service>                                                   \
  void ServiceProtocolServer<ChannelType>::Slot##n(                            \
      const RequestToken<Service>& request,                                    \
      const typename Service::Request::Slot& slot BOOST_PP_COMMA_IF(n)         \
      BOOST_PP_REPEAT(n, DECLARE_PARAMETER,)) {                                \
    AddPendingTask(static_cast<ChannelType*>(                                  \
      request.GetProtocol()->GetChannel()), boost::bind(slot, request          \
      BOOST_PP_COMMA_IF(n) BOOST_PP_REPEAT(n, LIST_PARAMETERS,)));             \
  }

  #define BOOST_PP_LOCAL_LIMITS (0, AVALON_SERVICE_PARAMETERS)
  #include BOOST_PP_LOCAL_ITERATE()
  #undef DECLARE_PARAMETER
  #undef PLACEHOLDERS
  #undef LIST_PARAMETERS
}
}

#endif // AVALON_SERVICEPROTOCOLSERVER_HPP
